/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.curator.framework.recipes.locks;

import com.google.common.base.Function;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import org.apache.curator.RetryLoop;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.utils.PathUtils;
import org.apache.curator.utils.ZKPaths;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;

import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 分布式锁（基于Zookeeper）
 * 互斥锁
 * 公平锁（监听上一临时顺序节点 + wait() / notifyAll()）
 * 可重入
 */
public class LockInternals {
    private final CuratorFramework client;
    // basePath + lockName
    private final String path;
    // 分布式锁指定的path
    private final String basePath;
    private final LockInternalsDriver driver;
    private final String lockName;
    private final AtomicReference<RevocationSpec> revocable = new AtomicReference<RevocationSpec>(null);
    private final CuratorWatcher revocableWatcher = new CuratorWatcher() {
        @Override
        public void process(WatchedEvent event) throws Exception {
            if (event.getType() == Watcher.Event.EventType.NodeDataChanged) {
                checkRevocableWatcher(event.getPath());
            }
        }
    };

    private final Watcher watcher = new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            notifyFromWatcher();
        }
    };

    private volatile int maxLeases;

    static final byte[] REVOKE_MESSAGE = "__REVOKE__".getBytes();

    /**
     * Attempt to delete the lock node so that sequence numbers get reset
     *
     * @throws Exception errors
     */
    public void clean() throws Exception {
        try {
            client.delete().forPath(basePath);
        } catch (KeeperException.BadVersionException ignore) {
            // ignore - another thread/process got the lock
        } catch (KeeperException.NotEmptyException ignore) {
            // ignore - other threads/processes are waiting
        }
    }

    /**
     * 最终构造函数。验证路径，组合路径 path + lockName
     */
    LockInternals(CuratorFramework client, LockInternalsDriver driver, String path, String lockName, int maxLeases) {
        this.driver = driver;
        this.lockName = lockName;
        this.maxLeases = maxLeases;
        PathUtils.validatePath(path);

        this.client = client;
        this.basePath = path;
        this.path = ZKPaths.makePath(path, lockName);
    }

    synchronized void setMaxLeases(int maxLeases) {
        this.maxLeases = maxLeases;
        notifyAll();
    }

    void makeRevocable(RevocationSpec entry) {
        revocable.set(entry);
    }

    void releaseLock(String lockPath) throws Exception {
        revocable.set(null);
        deleteOurPath(lockPath);
    }

    CuratorFramework getClient() {
        return client;
    }

    public static Collection<String> getParticipantNodes(CuratorFramework client, final String basePath,
                                                         String lockName, LockInternalsSorter sorter) throws Exception {
        List<String> names = getSortedChildren(client, basePath, lockName, sorter);
        Iterable<String> transformed = Iterables.transform
            (
                names,
                new Function<String, String>() {
                    @Override
                    public String apply(String name) {
                        return ZKPaths.makePath(basePath, name);
                    }
                }
            );
        return ImmutableList.copyOf(transformed);
    }

    public static List<String> getSortedChildren(CuratorFramework client, String basePath, final String lockName,
                                                 final LockInternalsSorter sorter) throws Exception {
        List<String> children = client.getChildren().forPath(basePath);
        List<String> sortedList = Lists.newArrayList(children);
        Collections.sort
            (
                sortedList,
                new Comparator<String>() {
                    @Override
                    public int compare(String lhs, String rhs) {
                        return sorter.fixForSorting(lhs, lockName).compareTo(sorter.fixForSorting(rhs, lockName));
                    }
                }
            );
        return sortedList;
    }

    public static List<String> getSortedChildren(final String lockName, final LockInternalsSorter sorter,
                                                 List<String> children) {
        List<String> sortedList = Lists.newArrayList(children);
        Collections.sort
            (
                sortedList,
                new Comparator<String>() {
                    @Override
                    public int compare(String lhs, String rhs) {
                        return sorter.fixForSorting(lhs, lockName).compareTo(sorter.fixForSorting(rhs, lockName));
                    }
                }
            );
        return sortedList;
    }

    List<String> getSortedChildren() throws Exception {
        return getSortedChildren(client, basePath, lockName, driver);
    }

    String getLockName() {
        return lockName;
    }

    LockInternalsDriver getDriver() {
        return driver;
    }

    String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception {
        final long startMillis = System.currentTimeMillis();
        // 无限等待时，millisToWait为null
        final Long millisToWait = (unit != null) ? unit.toMillis(time) : null;
        // 创建ZNode时节点存储的内容，默认为null
        final byte[] localLockNodeBytes = (revocable.get() != null) ? new byte[0] : lockNodeBytes;
        // 重试次数， 与CuratorFramework的重试策略有关
        int retryCount = 0;

        // 在Zookeeper中创建的临时顺序节点的路径，相当于一把待激活的分布式锁
        // 激活条件：同级目录子节点，名称排序最小（排队，公平锁）
        String ourPath = null;
        // 是否已经持有分布式锁
        boolean hasTheLock = false;
        // 是否已经完成尝试获取分布式锁的操作
        boolean isDone = false;
        while (!isDone) {
            isDone = true;

            try {
                if (localLockNodeBytes != null) {

                    /**
                     * 如果Client在获取到Server返回的有序节点的名称之前，Server崩溃或者Client连接断开了，
                     * 那么Client将不能知道创建的节点的名称，这时可能Client需要重新发起调用，为了解决这个问题，
                     * CreateBuilder提供了 withProtection方法， 来告诉Curator客户端用一个唯一标识符作为有序节点的前缀，
                     * 当Client创建有序节点失败，重试的时候， 就能根据该标识符判断有序节点是否创建成功。
                     */
                    // 创建一个临时有序的即诶单
                    ourPath = client.create().creatingParentsIfNeeded().withProtection().withMode(
                        CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path, localLockNodeBytes);
                } else {
                    ourPath = client.create().creatingParentsIfNeeded().withProtection().withMode(
                        CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path);
                }
                hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);
            } catch (KeeperException.NoNodeException e) {
                // 当标准lockinternalsdriver找不到锁节点时被抛出
                // 这可能在会话过期时发生，等等。所以，如果重试允许，就再试一次
                if (client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++,
                    System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper())) {
                    isDone = false;
                } else {
                    throw e;
                }
            }
        }

        if (hasTheLock) {
            return ourPath;
        }

        return null;
    }

    private void checkRevocableWatcher(String path) throws Exception {
        RevocationSpec entry = revocable.get();
        if (entry != null) {
            try {
                byte[] bytes = client.getData().usingWatcher(revocableWatcher).forPath(path);
                if (Arrays.equals(bytes, REVOKE_MESSAGE)) {
                    entry.getExecutor().execute(entry.getRunnable());
                }
            } catch (KeeperException.NoNodeException ignore) {
                // ignore
            }
        }
    }

    private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) throws Exception {
        boolean haveTheLock = false;
        boolean doDelete = false;
        try {
            if (revocable.get() != null) {
                client.getData().usingWatcher(revocableWatcher).forPath(ourPath);
            }

            // 状态为STARTED 且 未获取锁
            while ((client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock) {
                // 获取排序后的子节点列表
                List<String> children = getSortedChildren();
                // 获取前面自己创建的临时顺序子节点的名称
                String sequenceNodeName = ourPath.substring(basePath.length() + 1); // +1 包含/
                // 实现锁的公平性的核心逻辑
                PredicateResults predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);
                if (predicateResults.getsTheLock()) {
                    // 获得了锁，中断循环，继续返回上层
                    haveTheLock = true;
                } else {
                    // 没有获得到锁，监听上一临时顺序节点
                    String previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();

                    synchronized (this) {
                        try {
                            // use getData() instead of exists() to avoid leaving unneeded watchers which is a type
                            // of resource leak
                            // 上一临时顺序节点如果被删除，会唤醒当前线程继续竞争锁，正常情况下能直接获得锁，因为锁是公平的
                            client.getData().usingWatcher(watcher).forPath(previousSequencePath);
                            if (millisToWait != null) {
                                millisToWait -= (System.currentTimeMillis() - startMillis);
                                startMillis = System.currentTimeMillis();
                                if (millisToWait <= 0) {
                                    doDelete = true;    // 获取锁超时，标记删除之前创建的临时顺序节点
                                    break;
                                }

                                wait(millisToWait);
                            } else {
                                wait();
                            }
                        } catch (KeeperException.NoNodeException e) {
                            // it has been deleted (i.e. lock released). Try to acquire again
                        }
                    }
                }
            }
        } catch (Exception e) {
            // 标记删除，在finally删除之前创建的临时顺序节点（后台不断尝试）
            doDelete = true;
            // 重新抛出，尝试重新获取锁
            throw e;
        } finally {
            if (doDelete) {
                deleteOurPath(ourPath);
            }
        }
        return haveTheLock;
    }

    private void deleteOurPath(String ourPath) throws Exception {
        try {
            client.delete().guaranteed().forPath(ourPath);
        } catch (KeeperException.NoNodeException e) {
            // ignore - already deleted (possibly expired session, etc.)
        }
    }

    /**
     * 唤醒所有等待LockInternals实例的线程
     */
    private synchronized void notifyFromWatcher() {
        notifyAll();
    }
}
